package com.atguigu.userprofile.bitmap.app

import com.atguigu.userprofile.bean.TagInfo
import com.atguigu.userprofile.constant.ConstCode
import com.atguigu.userprofile.dao.TagInfoDAO
import com.atguigu.userprofile.util.ClickHouseUtil
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer


/**
 * 将每天跑成功的标签任务结果(Hive --> ClickHouse) 标签宽表,在ck中创建 bitmap结构的表, 并通过查询用户宽表的方式,将数据插入到 bitmap表中.
 */
object TaskBitmapApp {

  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setAppName("task_bitmap_app")//.setMaster("local[*]")

    val sparkContext = new SparkContext(sparkConf)
    /** 启动程序时,需要传入一个时间类型的日期,日期格式: 2020-06-14 */
    val taskDate: String = args(1)

    /*
        1、查询每天已经启用的标签任务列表,获取三级标签的表名.
     */
    val tagInfoList: List[TagInfo] = TagInfoDAO.getTagInfoListOnTask()


    /*
      2、根据 不同的标签值类型 存放在不同的列表中,这里主要做的工作: 根据存入的标签任务计算结果 tag_value的类型,创建多张表,每张表存的结果类型需要相同.
        例如:计算用户性别的标签任务,查询结果表结果: 男 | 女, 则存入到 `user_tag_value_string`中.查询性别的结果为 男 | 女 String类型的表中.
        查询最近30天下单件数的用户,查询结果为 Int值,则存入到 `user_tag_value_int`表中.
     */
    val tagInfoStringlist: ListBuffer[TagInfo] = new ListBuffer[TagInfo]();
    val tagInfoLonglist: ListBuffer[TagInfo] = new ListBuffer[TagInfo]();
    val tagInfoDecimallist: ListBuffer[TagInfo] = new ListBuffer[TagInfo]();
    val tagInfoDatelist: ListBuffer[TagInfo] = new ListBuffer[TagInfo]();

    for (tagInfo <- tagInfoList) {
      // tag_info建表语句中: tag_value_type  varchar(20)    null comment '1 整数 2 浮点 3 文本 4 日期',
      tagInfo.tagValueType match {
        case ConstCode.TAG_VALUE_TYPE_STRING => tagInfoStringlist.append(tagInfo)
        case ConstCode.TAG_VALUE_TYPE_DECIMAL => tagInfoDecimallist.append(tagInfo)
        case ConstCode.TAG_VALUE_TYPE_LONG => tagInfoLonglist.append(tagInfo)
        case ConstCode.TAG_VALUE_TYPE_DATE => tagInfoDatelist.append(tagInfo)
      }
    }
    /*
      3、把四个标签列表对应的标签数据 存储到4张bitmap表中,这里定义一个方法,insertSQLbyTagType(表名,List<三级标签名称>,任务执行时间).
        建表语句如下,相同标签任务计算结果对应的 uid,存储到 bitmap列中.并按照日期 + 标签名称分区.
        create table user_tag_value_string
        (
            tag_code  String,
            tag_value String,
            us        AggregateFunction(groupBitmap, UInt64),
            dt        String
        ) engine = AggregatingMergeTree()
              partition by (dt, tag_code)
              order by (tag_value);
     */
    insertSQLbyTagType("user_tag_value_string", tagInfoStringlist, taskDate)
    insertSQLbyTagType("user_tag_value_long", tagInfoLonglist, taskDate)
    insertSQLbyTagType("user_tag_value_decimal", tagInfoDecimallist, taskDate)
    insertSQLbyTagType("user_tag_value_date", tagInfoDatelist, taskDate)
  }

  /*
      INSERT INTO user_tag_value_string -- 表名需要替换成变量
      SELECT
         tag.1 AS tag_code,
         tag.2 AS tag_value,
         groupBitmapState(uid) AS us
      FROM
      (
         SELECT
             uid,
             arrayJoin([('agegroup', agegroup), ('gender', gender), ('favor', favor)]) AS tag -- 数组中[] 括号部分的内容,需要变量替换.并且要从List中取,第一个参数需要 ''拼接,第二个参数是具体的三级标签名称.
         FROM user_tag_merge
      ) AS ut
      GROUP BY tag.1,tag.2
   */
  def insertSQLbyTagType(tableName: String, tagList: ListBuffer[TagInfo], taskDate: String): Unit = {
    /** 在查询数据并插入当前bitmap结构的表前,先把各个表全部清理,主要考虑的是幂等操作. */
    /** alter table $tableName delete  where dt='$taskDate'  ; */
    val dropTableSql = s" alter table $tableName delete  where dt='$taskDate'"
    println(dropTableSql)
    ClickHouseUtil.executeSql(dropTableSql)


    if (tagList.size > 0) {
      // 取出 tagInfoStringlist中存的所有 tagInfo对象,将每个对象中的tagCode属性转化小写,并使用逗号分割,形成要查询的字段名. 拼接格式 ('字段名',字段值)
      /** [ =====> ('agegroup', agegroup), ('gender', gender), ('favor', favor) <====] */
      val queryCKWideTableTagCodesBySql = tagList.map(tagInfo => s"('${tagInfo.tagCode.toLowerCase}',${tagInfo.tagCode.toLowerCase})").mkString(",")
      /** ('tag_natural_characteristic_gender',tag_natural_characteristic_gender),('tag_natural_characteristic_agegroup',tag_natural_characteristic_agegroup) */
      println(queryCKWideTableTagCodesBySql)
      /** 在查询宽表,并插入到bitmap结构的表中时,查询结果为空的列,不能插入到 bitmap结构的宽表中,所以在查询时,加入了 where tag.2 <>'' 这个条件. */
      val insertSQL =
        s"""
           |    INSERT INTO $tableName SELECT
           |      tag.1 AS tag_code,
           |      tag.2 AS tag_value,
           |      groupBitmapState(cast( uid as UInt64)) AS us,
           |      '$taskDate'
           |   FROM
           |   (
           |      SELECT
           |           uid,
           |          arrayJoin([ $queryCKWideTableTagCodesBySql]) AS tag
           |      FROM user_tag_merge_${taskDate.replace("-", "")}
           |   ) AS ut
           |   where tag.2 <>''
           |  GROUP BY  tag.1,  tag.2
       """.stripMargin

      println(insertSQL)
      ClickHouseUtil.executeSql(insertSQL)
    }
  }
}
